package cc.xpcas.raft.context;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.*;
import java.util.stream.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cc.xpcas.raft.Configuration;
import cc.xpcas.raft.LogData;
import cc.xpcas.raft.LogEntry;
import cc.xpcas.raft.component.MultiIteratorMixer;
import cc.xpcas.raft.message.MessageContext;
import cc.xpcas.raft.network.Address;
import cc.xpcas.raft.network.Endpoint;
import cc.xpcas.raft.network.Poller;
import cc.xpcas.raft.role.Candidate;
import cc.xpcas.raft.role.Follower;
import cc.xpcas.raft.role.Leader;
import cc.xpcas.raft.role.Role;
import lombok.Getter;
import lombok.Setter;

/**
 * @author xp
 */
public class Context {

    private static final Logger LOG = LoggerFactory.getLogger(Context.class);

    @Getter
    private final Configuration configuration;

    public Context(Configuration configuration) {
        this.configuration = configuration;

        serverAddressIdMap = new HashMap<>();
        configuration.getPeers().forEach((serverId, address) -> {
            serverAddressIdMap.put(address, serverId);
        });
    }

    // ==== mainline thread ====

    @Getter
    private MainlineExecutor mainlineExecutor = new MainlineExecutor(this);

    public boolean isRunning() {
        return mainlineExecutor.isRunning();
    }

    public boolean isRunningOnMainline() {
        return mainlineExecutor.isRunningOnMainline();
    }

    public void runOnMainline(Runnable runnable) {
        mainlineExecutor.runOnMainline(runnable);
    }

    public void interruptMainline() {
        mainlineExecutor.interruptMainline();
    }

    public void enqueueMessage(MessageContext messageContext) {
        mainlineExecutor.enqueueMessage(messageContext);
    }

    // ==== scheduled thread ====

    private ScheduledExecutor scheduledExecutor = new ScheduledExecutor();

    public Future<?> schedule(Runnable origin, long delayInMillis) {
        return scheduledExecutor.schedule(origin, delayInMillis);
    }

    // ==== network ====

    @Getter
    @Setter
    private Endpoint endpoint;

    @Getter
    private final Map<Integer, Peer> peers = new HashMap<>();

    private final Map<Address, Integer> serverAddressIdMap;

    @Getter
    private final Poller poller = new Poller(this);

    public Peer getPeerById(int serverId) {
        return peers.get(serverId);
    }

    public Peer getPeerByAddress(Address address) {
        Integer serverId = serverAddressIdMap.get(address);
        if (serverId == null) return null;

        return getPeerById(serverId);
    }

    // ==== term ====

    private AtomicLong term = new AtomicLong(0L);

    public long getTerm() {
        return term.get();
    }

    public long incrementTerm() {
        return term.incrementAndGet();
    }

    /**
     * in mainline
     */
    public int catchUpTerm(long remoteTerm) {
        while (true) {
            long currentTerm = term.get();
            if (currentTerm > remoteTerm) return 1;
            if (currentTerm == remoteTerm) return 0;
            if (term.compareAndSet(currentTerm, remoteTerm)) {
                becomeFollower();
                votedFor = VOTED_FOR_NOBODY;
                return -1;
            }
        }
    }

    // ==== role ====

    public static final int VOTED_FOR_NOBODY = -1;

    @Getter
    @Setter
    private int votedFor;

    @Getter
    private Role role;

    /**
     * in mainline
     */
    public boolean isLeader() {
        return role instanceof Leader;
    }

    /**
     * in mainline
     */
    public boolean isCandidate() {
        return role instanceof Candidate;
    }

    /**
     * in mainline
     */
    public boolean isFollower() {
        return role instanceof Follower;
    }

    /**
     * in mainline
     */
    private <T extends Role> void becomeNewRole(Class<T> roleClass, Function<Context, T> constructor) {
        if (roleClass.isInstance(role)) {
            // no change
            return;
        }
        if (role != null) {
            try {
                role.end();
            } catch (InterruptedException e) {
                mainlineExecutor.close();
                role = null;
                return;
            }
        }
        role = constructor.apply(this);
        runOnMainline(role::start);
    }

    /**
     * in mainline
     */
    public void becomeCandidate() {
        becomeNewRole(Candidate.class, Candidate::new);
    }

    /**
     * in mainline
     */
    public void becomeLeader() {
        becomeNewRole(Leader.class, Leader::new);
    }

    /**
     * in mainline
     */
    public void becomeFollower() {
        becomeNewRole(Follower.class, Follower::new);
    }

    // ==== log && 2-phase commit ====

    private List<LogEntry> preparedLogs;

    private Deque<LogEntry> preparedLogsQueue;

    {
        LinkedList<LogEntry> linkedList = new LinkedList<>();
        preparedLogs = linkedList;
        preparedLogsQueue = linkedList;
    }

    private List<LogEntry> committedLogs;

    private Deque<LogEntry> committedLogQueue;

    {
        LinkedList<LogEntry> linkedList = new LinkedList<>();
        committedLogs = linkedList;
        committedLogQueue = linkedList;
    }

    @Setter
    private Consumer<LogData> committer;

    /**
     * in mainline
     */
    public LogEntry getLogEntry(long logId) {
        return Stream.concat(
                committedLogs.parallelStream(),
                preparedLogs.parallelStream()
        ).filter(entry -> entry.getId() == logId).findAny().orElse(null);
    }

    /**
     * in mainline
     */
    public Iterator<LogEntry> getLogEntriesSince(long logId) {
        return new MultiIteratorMixer<>(
                committedLogs.iterator(),
                preparedLogs.iterator()
        ).skipUntil(entry -> entry.getId() >= logId);
    }

    /**
     * in mainline
     */
    public void appendLogEntries(Collection<LogEntry> entries) {
        if (entries.isEmpty()) return;

        Iterator<LogEntry> appendIterator = entries.iterator();
        LogEntry head = appendIterator.next();

        ListIterator<LogEntry> preparedIterator = preparedLogs.listIterator();
        while (preparedIterator.hasPrevious()) {
            LogEntry prepared = preparedIterator.previous();
            if (prepared.getId() < head.getId()) break;
        }
        // preparedIterator is the last entry lesser than head
        while (preparedIterator.hasNext()) {
            LogEntry prepared = preparedIterator.next();
            if (Objects.equals(prepared, head)) {
                if (appendIterator.hasNext()) {
                    // matched
                    head = appendIterator.next();
                } else {
                    // wired, the append is short than the prepared
                    return;
                }
            } else {
                dropLogSince(prepared.getId());
                break;
            }
        }
        // appendIterator is the first entry different
        while (appendIterator.hasNext()) {
            preparedLogs.add(appendIterator.next());
        }
    }

    /**
     * in mainline
     */
    private void dropLogSince(long logId) {
        while (true) {
            LogEntry entry = preparedLogsQueue.peekLast();
            if (entry == null) break;
            if (entry.getId() < logId) break;

            preparedLogsQueue.pollLast();
        }
    }

    /**
     * in mainline
     */
    public void catchUpCommittedLogId(long committedLogId) {
        while (true) {
            LogEntry entry = preparedLogsQueue.peekFirst();
            if (entry == null) break;
            if (entry.getId() > committedLogId) break;

            preparedLogsQueue.pollFirst();
            committedLogs.add(entry);
            committer.accept(entry.getData());
        }
    }

    /**
     * in mainline
     */
    public LogEntry getLastLog() {
        if (preparedLogsQueue.isEmpty()) {
            return committedLogQueue.peekLast();
        } else {
            return preparedLogsQueue.peekLast();
        }
    }

    /**
     * in mainline
     */
    public long getLastLogId() {
        LogEntry entry = getLastLog();
        if (entry == null) return 0L;
        return entry.getId();
    }

    /**
     * in mainline
     */
    public long getLastCommittedLogId() {
        LogEntry entry = committedLogQueue.peekLast();
        if (entry == null) return 0L;
        return entry.getId();
    }
}
